import concurrent.futures
import paho.mqtt.client as mqtt
import json
import time
import random

from utils.config import Config


class MQTTClient():
    def __init__(self):
        self.config = Config('./conf/config.ini')

        self.client = mqtt.Client(client_id=self.config.read_value('mqtt', 'client_id')+ str(random.randint(100000, 999999)))
        self.client.username_pw_set(self.config.read_value('mqtt', 'username'), self.config.read_value('mqtt', 'password'))
        self.client.on_connect = self.on_connect
        self.broker_address = self.config.read_value('mqtt', 'broker_address')
        self.port = int(self.config.read_value('mqtt', 'port'))

        self.client.connect(self.broker_address, self.port,keepalive=60000)  # 在初始化时建立连接

    def __del__(self):
        self.client.disconnect()  # 在销毁实例时断开连接

    def on_connect(self, client, userdata, flags, rc):
        if rc == 0:
            print("Connected to MQTT broker")
        else:
            print("Connection failed")

    # 发布信息
    def publish_message(self, topic, payload):
        payload_str = payload.decode('utf-8') if isinstance(payload, bytes) else payload
        self.client.publish(topic, json.dumps(payload_str, ensure_ascii=False))

    # 订阅
    def subscribe_topic(self, topic):
        send_subscribe=self.client.subscribe(topic)
        print(send_subscribe)
        self.client.loop_forever()

    # 发布多条消息
    def publish_multiple_messages(self, topic, payloads):
        for payload in payloads:
            payload_str = json.dumps(payload, ensure_ascii=False).encode('utf-8')  # 编码中文内容为UTF-8
            self.client.publish(topic, payload_str)
            time.sleep(1)  # 每条消息之间暂停一秒

    # 并发发布消息
    def concurrent_publish(self,topic,payloads):
        with concurrent.futures.ThreadPoolExecutor(max_workers=len(payloads)) as executor:  # 设置并发线程数
            future_to_thread = {executor.submit(self.publish_message, topic, payloads[i]): payloads[i] for i in range(len(payloads))}
            for future in concurrent.futures.as_completed(future_to_thread):
                thread_num = future_to_thread[future]
                try:
                    future.result()
                except Exception as exc:
                    print(f'Thread {thread_num} generated an exception: {exc}')

